package com.gitee.wsl.platform.thread.task.bean

import com.gitee.wsl.platform.exception.rethrowIfFatal
import com.gitee.wsl.platform.thread.task.TaskCancellationException
import com.gitee.wsl.platform.thread.task.TaskExecutor
import com.gitee.wsl.platform.thread.task.TaskExecutors.sharedBlockingIO
import com.gitee.wsl.platform.thread.task.TaskLocalContext.signalTheStartOfBlockingCall
import com.gitee.wsl.platform.thread.task.Trampoline.execute
import com.gitee.wsl.platform.thread.task.api.AsyncFun
import com.gitee.wsl.platform.concurrent.api.Cancellable
import com.gitee.wsl.platform.thread.task.api.DelayedFun
import com.gitee.wsl.platform.thread.task.api.ExecutedFiber
import com.gitee.wsl.platform.thread.task.api.Fiber
import com.gitee.wsl.platform.thread.task.callback.AsyncContinuationCallback
import com.gitee.wsl.platform.thread.task.callback.AsyncContinuationFun
import com.gitee.wsl.platform.thread.task.callback.BlockingCompletionCallback
import com.gitee.wsl.platform.thread.task.callback.CancellableContinuation
import com.gitee.wsl.platform.thread.task.callback.CompletionCallback
import com.gitee.wsl.platform.thread.task.callback.Continuation
import org.jetbrains.annotations.ApiStatus
import org.jetbrains.annotations.Blocking
import org.jetbrains.annotations.NonBlocking
import java.time.Duration
import java.util.concurrent.*


/**
 * Represents a function that can be executed asynchronously.
 */
class Task<T> private constructor(
    private val createFun: AsyncContinuationFun<T>
) {

    /**
     * Ensure that the task will start execution on the injected [Executor].
     *
     *
     * The execution model is set by the "run" methods, e.g.,
     * [.runAsync] will create a new thread,
     * but [.runBlocking] will not, trying to execute the task
     * on the current thread. Thus, it may be useful to instruct the task to
     * always run on a thread managed by the given executor.
     *
     * <pre>`Task.fromBlockingIO(() -> {
     * // Reads a file from disk
     * return Files.readString(Paths.get("file.txt"));
     * }).ensureRunningOnExecutor().runBlocking(
     * TaskExecutors.sharedBlockingIO()
     * );
    `</pre> *
     *
     * In the above example we are reading a file from disk. Normally, the
     * execution via `runBlocking` would start this task on the current
     *
     */
    @JvmOverloads
    fun ensureRunningOnExecutor(executor: Executor? = null): Task<T> {
        return Task(AsyncContinuationFun { cont ->
            val cont2: Continuation<T> = if (executor != null)
                cont.withExecutorOverride(TaskExecutor.from(executor))
            else
                cont
            cont2.executor.resumeOnExecutor{ createFun.invoke(cont2) }
        })
    }

    /**
     * Guarantees that the task will complete with the given callback.
     *
     *
     * This method is useful for releasing resources or performing
     * cleanup operations when the task completes.
     *
     *
     * This callback will be invoked in addition to whatever the client
     * provides as a callback to [.runAsync]
     * or similar methods.
     *
     *
     * **WARNING:** The invocation of this method is concurrent
     * with the task's completion, meaning that ordering isn't guaranteed
     * (i.e., a callback installed with this method may be called before or
     * after the callback provided to [.runAsync]).
     */
    fun withOnComplete(callback: CompletionCallback<T>): Task<T> {
        return Task(AsyncContinuationFun { cont ->
            cont.registerExtraCallback(callback)
            cont.executor.resumeOnExecutor { createFun.invoke(cont) }
        })
    }

    /**
     * Registers a [Cancellable] that can be used to cancel the running task.
     */
    fun withCancellation(
        cancellable: Cancellable,
    ): Task<T> {
        return Task(AsyncContinuationFun { cont ->
            cont.registerCancellable(cancellable)
            cont.executor.resumeOnExecutor { createFun.invoke(cont) }
        })
    }

    /**
     * Executes the task asynchronously.
     *
     *
     * This method ensures that the start starts execution on a different thread,
     * managed by the given executor.
     *
     * @param callback will be invoked when the task completes
     * @param executor is the [Executor] that may be used to run the task
     * @return a [Cancellable] that can be used to cancel a running task
     */
    fun runAsync(
        executor: Executor ?= null ,
        callback: CompletionCallback<T>,
    ): Cancellable {
        val executor = executor?:sharedBlockingIO()
        val taskExecutor = TaskExecutor.from(executor)
        val cont = CancellableContinuation(
            taskExecutor,
            AsyncContinuationCallback(
                callback,
                taskExecutor
            )
        )
        taskExecutor.execute {
            try {
                createFun.invoke(cont)
            } catch (e: Throwable) {
                e.rethrowIfFatal()
                cont.onFailure(e)
            }
        }
        return cont
    }

    /**
     * Overload of [.runAsync] that
     * uses [TaskExecutors.sharedBlockingIO] as the executor.
     *
     *
     * Starts asynchronous execution, on a different thread.
     *
     * @param callback will be invoked when the task completes
     * @return a [Cancellable] that can be used to cancel a running task
     */
//    fun runAsync(callback: CompletionCallback<T>): Cancellable {
//        return runAsync(null, callback)
//    }

    /**
     * Executes the task concurrently and returns a [Fiber] that can be
     * used to wait for the result or cancel the task.
     *
     *
     * Similar to [.runAsync], this method
     * starts the execution on a different thread.
     *
     * @param executor is the [Fiber] that may be used to run the task
     * @return a [Fiber] that can be used to wait for the outcome,
     * or to cancel the running fiber.
     */
    @NonBlocking
    fun runFiber(executor: Executor? = null ): Fiber<T> {
        return ExecutedFiber.start(executor?:sharedBlockingIO() ,createFun)
    }

    /**
     * Overload of [.runFiber] that
     * uses [TaskExecutors.sharedBlockingIO] as the executor.
     *
     *
     * Similar to [.runAsync], this method
     * starts the execution on a different thread.
     *
     * @return a [Fiber] that can be used to wait for the outcome,
     * or to cancel the running fiber.
     */
//    @NonBlocking
//    fun runFiber(): Fiber<T> {
//        return runFiber(null)
//    }

    /**
     * Executes the task and blocks until it completes, or the current
     * thread gets interrupted (in which case the task is also cancelled).
     *
     *
     * Given that the intention is to block the current thread for the result,
     * the task starts execution on the current thread.
     *
     * @param executor is the [Executor] that may be used to run the task
     * @return the successful result of the task
     *
     * @throws ExecutionException if the task fails with an exception
     *
     * @throws InterruptedException if the current thread is interrupted, which
     * also cancels the running task. Note that on interruption, the running
     * concurrent task must also be interrupted, as this method always blocks
     * for its interruption or completion.
     */
    @Blocking
    @Throws(ExecutionException::class, InterruptedException::class)
    fun runBlocking(executor: Executor? = null): T {
        val executor = executor?:sharedBlockingIO()
        val blockingCallback = BlockingCompletionCallback<T>()
        val taskExecutor = TaskExecutor.from(executor)
        val cont = CancellableContinuation(taskExecutor, blockingCallback)
        createFun.invoke(cont)
        signalTheStartOfBlockingCall()
        return blockingCallback.await(cont)
    }

    /**
     * Overload of [.runBlocking] that uses
     * [TaskExecutors.sharedBlockingIO] as the executor.
     *
     *
     * Given that the intention is to block the current thread for the result,
     * the task starts execution on the current thread.
     *
     * @return the successful result of the task
     *
     * @throws ExecutionException if the task fails with an exception
     *
     * @throws InterruptedException if the current thread is interrupted, which
     * also cancels the running task. Note that on interruption, the running
     * concurrent task must also be interrupted, as this method always blocks
     * for its interruption or completion.
     */
//    @Blocking
//    @Throws(ExecutionException::class, InterruptedException::class)
//    fun runBlocking(): T? {
//        return runBlocking(null)
//    }

    /**
     * Executes the task and blocks until it completes, or the timeout is reached,
     * or the current thread is interrupted.
     *
     *
     * **EXECUTION MODEL:** Execution starts on a different thread,
     * by necessity, otherwise the execution could block the current thread
     * indefinitely, without the possibility of interrupting the task after
     * the timeout occurs.
     *
     * @param timeout  is the maximum time to wait for the task to complete
     * before throwing a [TimeoutException]
     * @param executor is the [Executor] that may be used to run
     * the task
     * @return the successful result of the task
     * @throws ExecutionException   if the task fails with an exception
     * @throws InterruptedException if the current thread is interrupted.
     * The running task is also cancelled, and this method does not
     * return until `onCancel` is signaled.
     * @throws TimeoutException     if the task doesn't complete within the
     * specified timeout. The running task is also cancelled on timeout,
     * and this method does not returning until `onCancel` is signaled.
     */
    @Throws(ExecutionException::class, InterruptedException::class, TimeoutException::class)
    fun runBlockingTimed(
        executor: Executor?=null,
        timeout: Duration,
    ): T? {
        val executor = executor?:sharedBlockingIO()
        val blockingCallback = BlockingCompletionCallback<T>()
        val taskExecutor = TaskExecutor.from(executor )
        val cont = CancellableContinuation(taskExecutor, blockingCallback)
        taskExecutor.execute({ createFun.invoke(cont) })
        return blockingCallback.await(cont, timeout)
    }

    /**
     * Overload of [.runBlockingTimed] that
     * uses [TaskExecutors.sharedBlockingIO] as the executor.
     *
     *
     * **EXECUTION MODEL:** Execution starts on a different thread,
     * by necessity, otherwise the execution could block the current thread
     * indefinitely, without the possibility of interrupting the task after
     * the timeout occurs.
     *
     * @param timeout is the maximum time to wait for the task to complete
     * @return the successful result of the task
     * @throws ExecutionException   if the task fails with an exception
     * @throws InterruptedException if the current thread is interrupted.
     * The running task is also cancelled, and this method does not
     * return until `onCancel` is signaled.
     * @throws TimeoutException     if the task doesn't complete within the
     * specified timeout. The running task is also cancelled on timeout,
     * and this method does not returning until `onCancel` is signaled.
     */
//    @Throws(ExecutionException::class, InterruptedException::class, TimeoutException::class)
//    fun runBlockingTimed(timeout: Duration): T? {
//        return runBlockingTimed(null, timeout)
//    }

    companion object {
        /**
         * Creates a task from an asynchronous computation, initiated on the current
         * thread.
         *
         *
         * This method ensures:
         *
         *  1. Idempotent cancellation
         *  1. Trampolined execution to avoid stack-overflows
         *
         *
         *
         *
         * @param start is the function that will trigger the async computation,
         * injecting a callback that will be used to signal the result,
         * and an executor that can be used for creating additional threads.
         * @return a new task that will execute the given builder function upon execution
         */
        fun <T> fromAsync(start: AsyncFun<out T>): Task<T> {
            return Task(AsyncContinuationFun { cont ->
                execute(Runnable {
                    try {
                        cont.registerForwardCancellable().set(
                            start.invoke(cont.executor, cont)
                        )
                    } catch (e: Throwable) {
                        e.rethrowIfFatal()
                        cont.onFailure(e)
                    }
                })
            })
        }

        /**
         * Creates a task from a [DelayedFun] executing blocking IO.
         *
         *
         * This uses Java's interruption protocol (i.e., [Thread.interrupt])
         * for cancelling the task.
         */
        fun <T> fromBlockingIO(run: DelayedFun<out T>): Task<T> {
            return Task(AsyncContinuationFun { cont ->
                val th = Thread.currentThread()
                val registration = cont.registerCancellable({ th.interrupt() })
                if (registration == null) {
                    cont.onCancellation()
                    return@AsyncContinuationFun
                }
                try {
                    var result: T?
                    try {
                        signalTheStartOfBlockingCall()
                        result = run.invoke()
                    } finally {
                        registration.cancel()
                    }
                    if (th.isInterrupted) {
                        throw InterruptedException()
                    }
                    cont.onSuccess(result)
                } catch (e: InterruptedException) {
                    cont.onCancellation()
                } catch (e: TaskCancellationException) {
                    cont.onCancellation()
                } catch (e: Throwable) {
                    e.rethrowIfFatal()
                    cont.onFailure(e)
                } finally {
                    Thread.interrupted()
                }
            })
        }

        /**
         * Creates a task from a [Future] builder.
         *
         *
         * This is compatible with Java's interruption protocol and
         * [Future.cancel], with the resulting task being cancellable.
         *
         *
         * **NOTE:** Use [.fromCompletionStage] for directly
         * converting [CompletableFuture] builders, because it is not possible to cancel
         * such values, and the logic needs to reflect it. Better yet, use
         * [.fromCancellableFuture] for working with [CompletionStage]
         * values that can be cancelled.
         *
         * @param builder is the [DelayedFun] that will create the [Future] upon
         * this task's execution.
         * @return a new task that will complete with the result of the created `Future`
         * upon execution
         * @see .fromCompletionStage
         * @see .fromCancellableFuture
         */
        fun <T > fromBlockingFuture(builder: DelayedFun<Future<out T>>): Task<T> {
            return fromBlockingIO(DelayedFun {
                val f = builder.invoke()
                try {
                    signalTheStartOfBlockingCall()
                    return@DelayedFun f.get()
                } catch (e: ExecutionException) {
                    if (e.cause is RuntimeException) throw e.cause as RuntimeException
                    throw e
                } catch (e: InterruptedException) {
                    f.cancel(true)
                    // We need to wait for this future to complete, as we need to
                    // try to back-pressure on its interruption (doesn't really work).
                    while (!f.isDone) {
                        // Ignore further interruption signals
                        Thread.interrupted()
                        try {
                            f.get()
                        } catch (ignored: Exception) {
                        }
                    }
                    throw e
                }
            })
        }

        /**
         * Creates tasks from a builder of [CompletionStage].
         *
         *
         * **NOTE:** `CompletionStage` isn't cancellable, and the
         * resulting task should reflect this (i.e., on cancellation, the listener should not
         * receive an `onCancel` signal until the `CompletionStage` actually completes).
         *
         *
         * Prefer using [.fromCancellableFuture] for working with
         * [CompletionStage] values that can be cancelled.
         *
         * @param builder is the [DelayedFun] that will create the [CompletionStage]
         * value. It's a builder because [Task] values are cold values
         * (lazy, not executed yet).
         * @return a new task that upon execution will complete with the result of
         * the created `CancellableCompletionStage`
         * @see .fromCancellableFuture
         */
        fun <T> fromCompletionStage(
            builder: DelayedFun<out CompletionStage<out T>>,
        ): Task<T> {
            return fromCancellableFuture(
                DelayedFun {
                    CancellableFuture(
                        builder.invoke().toCompletableFuture(),
                        Cancellable.EMPTY
                    )
                }
            )
        }

        /**
         * Creates tasks from a builder of [CancellableFuture].
         *
         *
         * This is the recommended way to work with [CompletionStage] builders,
         * because cancelling such values (e.g., [CompletableFuture]) doesn't work
         * for cancelling the connecting computation. As such, the user should provide
         * an explicit [Cancellable] token that can be used.
         *
         * @param builder is the [DelayedFun] that will create the [CancellableFuture]
         * value. It's a builder because [Task] values are cold values
         * (lazy, not executed yet).
         * @return a new task that upon execution will complete with the result of
         * the created [CancellableFuture]
         */
        fun <T> fromCancellableFuture(
            builder: DelayedFun<CancellableFuture<out T>>,
        ): Task<T> {
            return Task(TaskFromCancellableFuture(builder))
        }

        /**
         * Creates a task that completes with the given static/pure value.
         */
        fun <T> pure(value: T): Task<T> {
            return Task(AsyncContinuationFun { cont -> cont.onSuccess(value) })
        }

        /** Reusable "void" task that does nothing, completing immediately.  */
        val NOOP: Task<Void?> = Task(AsyncContinuationFun { cont -> cont.onSuccess(null) })
    }
}

/**
 * INTERNAL API.
 *
 *
 * **INTERNAL API:** Internal apis are subject to change or removal
 * without any notice. When code depends on internal APIs, it is subject to
 * breakage between minor version updates.
 */
@ApiStatus.Internal
internal class TaskFromCancellableFuture<T>(
    private val builder: DelayedFun<CancellableFuture<out T>>
) : AsyncContinuationFun<T> {

    public override fun invoke(continuation: Continuation<T>) {
        try {
            val cancellableRef = continuation.registerForwardCancellable()
            val cancellableFuture = builder.invoke()
            val future = getCompletableFuture(cancellableFuture, continuation)
            val cancellable: Cancellable = cancellableFuture.cancellable

            cancellableRef.set({
                try {
                    cancellable.cancel()
                } finally {
                    future.cancel(true)
                }
            })
        } catch (e: Throwable) {
            e.rethrowIfFatal()
            continuation.onFailure(e)
        }
    }

    companion object {
        private fun <T> getCompletableFuture(
            cancellableFuture: CancellableFuture<out T>,
            callback: CompletionCallback<in T>,
        ): CompletableFuture<out T> {
            val future = cancellableFuture.future
            future.whenComplete { value: T, error: Throwable? ->
                if (error is InterruptedException || error is TaskCancellationException) {
                    callback.onCancellation()
                } else if (error is ExecutionException) {
                    val e = if (error.cause != null) error.cause else error
                    callback.onFailure(e!!)
                } else if (error is CompletionException) {
                    val e = if (error.cause != null) error.cause else error
                    callback.onFailure(e!!)
                } else {
                    if (error != null) callback.onFailure(error)
                    else callback.onSuccess(value)
                }
            }
            return future
        }
    }
}
